package com.by.hazelcasttopic;

import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson2.JSON;
import com.by.hazelcasttopic.annotation.HazelcastTopicListener;
import com.by.hazelcasttopic.properties.HazelcastTopicProperties;
import com.by.toolkitspring.customizeannotation.ByScan;
import com.hazelcast.cluster.*;
import com.hazelcast.config.Config;
import com.hazelcast.config.JoinConfig;
import com.hazelcast.config.NetworkConfig;
import com.hazelcast.config.RestEndpointGroup;
import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.topic.ITopic;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.jetbrains.annotations.NotNull;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;

import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;

@Slf4j
@ComponentScan({"com.by.hazelcasttopic"})
@Configuration
@ConditionalOnProperty(name = "by.hazelcasttopic.enabled", havingValue = "true")
@EnableConfigurationProperties({HazelcastTopicProperties.class})
@ByScan(basePackage = "com.by", annotationClass = HazelcastTopicListener.class)
public class HazelcastTopicAutoConfiguration {

    @Resource
    private HazelcastTopicProperties hazelcastTopicProperties;

    @Resource
    private GlobalHazelcastTopicListener listener;

    public static final String SERVICE_DNS = "{}.{}.svc.cluster.local";

    @Bean
    public HazelcastTopic hazelcastConfiguration() {
        String k8sNameSpace = hazelcastTopicProperties.getK8sNameSpace();
        if (StringUtils.isBlank(k8sNameSpace)) {
            throw new RuntimeException("k8s集群命名空间不能为空！");
        }
        Map<String, HazelcastTopicProperties.Config> clusterMap = hazelcastTopicProperties.getClusterMap();
        if (Objects.isNull(clusterMap) || clusterMap.isEmpty()) {
            throw new RuntimeException("集群信息不能为空！");
        }
        log.info("[HazelcastTopicAutoConfiguration][hazelcastConfiguration] k8sNameSpace:[{}] clusterMap:[{}]"
                , k8sNameSpace, clusterMap);

        HazelcastTopic hazelcastTopic = new HazelcastTopic();

        Map<String, HazelcastInstance> clusterInstanceMap = clusterMap.entrySet().stream()
                .collect(Collectors.toMap(Map.Entry::getKey
                        , clusterInfo -> {
                            String clusterKey = clusterInfo.getKey();
                            HazelcastTopicProperties.Config configInfo = clusterInfo.getValue();
                            int port = configInfo.getPort();
                            String serviceName = configInfo.getServiceName();
                            int servicePort = configInfo.getServicePort();
                            log.info("[HazelcastTopicAutoConfiguration][hazelcastConfiguration] clusterKey:[{}] configInfo:[{}]", clusterKey, configInfo);
                            Config config = new Config();
                            config.setClusterName(clusterKey)
                                    // 节点启动后等待30秒再尝试加入集群
                                    .setProperty("hazelcast.cluster.initial.wait.seconds", "30")
                                    .setProperty("hazelcast.discovery.enabled", "true")
                                    .setProperty("hazelcast.socket.bind.any", "false")
                                    .setProperty("hazelcast.socket.server.bind.any", "false");

                            config.getCPSubsystemConfig().setCPMemberCount(0);

                            // 配置网络
                            NetworkConfig networkConfig = config.getNetworkConfig();
                            networkConfig
                                    .setPort(port)
                                    .setPortAutoIncrement(false)
                                    .getRestApiConfig()
                                    .setEnabled(true)
                                    .enableGroups(RestEndpointGroup.HEALTH_CHECK);

                            // 配置集群发现（例如 Kubernetes）
                            JoinConfig joinConfig = networkConfig.getJoin();
                            joinConfig.getMulticastConfig().setEnabled(false); // 禁用多播
                            joinConfig.getTcpIpConfig().setEnabled(false); // 禁用 TCP/IP
                            joinConfig.getKubernetesConfig() // 启用 Kubernetes 发现
                                    .setEnabled(true)
                                    .setProperty("service-port", String.valueOf(servicePort))
                                    .setProperty("service-dns", StrUtil.format(SERVICE_DNS, serviceName, k8sNameSpace))
                                    .setProperty("service-dns-timeout", "5");
                            return Hazelcast.newHazelcastInstance(config);
                        }));

        clusterMap.forEach((clusterName, config) -> {
            HazelcastInstance hazelcastInstance = clusterInstanceMap.get(clusterName);
            List<String> topics = config.getTopics();
            // 获取集群成员
            Set<Member> members = this.listClusterMembers(hazelcastInstance);
            List<String> clusterHosts = members.stream()
                    .map(Member::getAddress)
                    .map(Address::getHost)
                    .collect(Collectors.toList());
            log.info("[HazelcastTopicAutoConfiguration][hazelcastConfiguration] clusterName:[{}] clusterMembersHosts:[{}]", clusterName, JSON.toJSONString(clusterHosts));

            Map<String, HazelcastInstance> hazelcastInstanceMap = topics.stream()
                    .collect(Collectors.toMap(Function.identity(), topicName -> {
                        ITopic<String> topic = hazelcastInstance.getTopic(topicName);
                        topic.addMessageListener(listener);
                        log.info("[HazelcastTopicAutoConfiguration][hazelcastConfiguration] addMessageListenerSuccess clusterName:[{}] topicName:[{}] listener:[{}]", clusterName, topicName, listener);
                        return hazelcastInstance;
                    }));
            log.info("[HazelcastTopicAutoConfiguration][hazelcastConfiguration] clusterName:[{}] hazelcastInstanceMap:[{}]", clusterName, hazelcastInstanceMap);
            hazelcastTopic.setInstances(hazelcastInstanceMap);
        });
        return hazelcastTopic;
    }

    @NotNull
    private Set<Member> listClusterMembers(HazelcastInstance hazelcastInstance) {
        Config config = hazelcastInstance.getConfig();
        Cluster cluster = hazelcastInstance.getCluster();
        String clusterName = config.getClusterName();
        cluster.addMembershipListener(new MembershipListener() {
            @Override
            public void memberAdded(MembershipEvent membershipEvent) {
                Member member = membershipEvent.getMember();
                Set<Member> members = membershipEvent.getMembers();
                log.info("[HazelcastTopicAutoConfiguration][memberAdded] clusterName:[{}] memberAddedHost:[{}] clusterMembersHosts:[{}]"
                        , clusterName
                        , member.getAddress().getHost()
                        , members.stream()
                                .map(Member::getAddress)
                                .map(Address::getHost)
                                .collect(Collectors.toList()));
            }

            @Override
            public void memberRemoved(MembershipEvent membershipEvent) {
                Member member = membershipEvent.getMember();
                Set<Member> members = membershipEvent.getMembers();
                log.info("[HazelcastTopicAutoConfiguration][memberRemoved] clusterName:[{}] memberRemovedHost:[{}] clusterMembersHosts:[{}]"
                        , clusterName
                        , member.getAddress().getHost()
                        , members.stream()
                                .map(Member::getAddress)
                                .map(Address::getHost)
                                .collect(Collectors.toList()));
            }
        });
        return cluster.getMembers();
    }


}
